package com.ydl.learning.flink.demo.mySink

import com.alibaba.fastjson.{JSON, JSONObject}
import com.ydl.learning.flink.demo.JavaUtils
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.http.HttpHost
import org.apache.http.auth.{AuthScope, UsernamePasswordCredentials}
import org.apache.http.client.CredentialsProvider
import org.apache.http.impl.client.BasicCredentialsProvider
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder
import org.elasticsearch.action.delete.DeleteRequest
import org.elasticsearch.action.get.GetRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.update.UpdateRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType

/**
 *
 *
 * @author ydl
 * @since 2021/10/21
 */
class EsUpdateSink extends RichSinkFunction[String] {
  private var client: RestHighLevelClient = _
  private val index = "test"

  override def invoke(value: String, context: SinkFunction.Context): Unit = {
    val json = new JSONObject()
    json.put("a", value)
    //添
    //val indexRequest = new IndexRequest(index)
    //indexRequest.id("1")
    //indexRequest.source(json.toJSONString, XContentType.JSON)
    //client.index(indexRequest, RequestOptions.DEFAULT)
    //改
    val updateRequest = new UpdateRequest(index,"1")
    updateRequest.doc(json.toJSONString, XContentType.JSON)
    client.update(updateRequest, RequestOptions.DEFAULT)
    //查
    //val getRequest = new GetRequest(index,"1")
    //println(JavaUtils.o2jsonstr(client.get(getRequest,RequestOptions.DEFAULT).getSourceAsString))
    //删
    //val delRequest = new DeleteRequest(index,"1")
    //client.delete(delRequest, RequestOptions.DEFAULT)
    println(value)
  }

  override def open(parameters: Configuration): Unit = {
    val host = new HttpHost("10.50.168.216", 9200, HttpHost.DEFAULT_SCHEME_NAME)
    val builder: RestClientBuilder = RestClient.builder(host)
    val credentialsProvider: CredentialsProvider = new BasicCredentialsProvider()
    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("", ""))
    val callback = new RestClientBuilder.HttpClientConfigCallback() {
      override def customizeHttpClient(httpAsyncClientBuilder: HttpAsyncClientBuilder): HttpAsyncClientBuilder = {
        httpAsyncClientBuilder.disableAuthCaching
        httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
      }
    }
    builder.setHttpClientConfigCallback(callback)
    client = new RestHighLevelClient(builder)
  }

  override def close(): Unit = {
    client.close()
  }
}
